package cn.qdhlm.collectservice.core.callback;


import cn.qdhlm.collectservice.common.util.CodeUtil;
import cn.qdhlm.collectservice.core.handler.CommandHandler;
import cn.qdhlm.collectservice.core.handler.DataHandler;
import cn.qdhlm.collectservice.model.rocketmq.DeviceData;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * @Name:RocketMqMessageCallback
 * @description:
 * @author: xinY
 **/
@Log4j2
@Component
public class RocketMqMessageCallback implements MessageListenerConcurrently {
    @Autowired
    private DataHandler dataHandler;
    @Autowired
    private CommandHandler commandHandler;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list) {
            log.info("topic:" + msg.getTopic()
                    + ", tags:" + msg.getTags()
                    + ", body:" + ByteBufUtil.hexDump(msg.getBody())
                    + ", from broker:" + msg.getBrokerName()
                    + ", properties:" + msg.getProperty("KEYS")
                    + ", from queue:" + msg.getQueueId()
                    + ", message ID:" + msg.getMsgId()

            );
            if ("DeviceCommand".equals(msg.getTopic())) {
                String result = commandHandler.process(msg);
                System.out.println("result::::::::"+result);
            } else if ("DeviceData".equals(msg.getTopic())) {
                dataHandler.process(msg);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
